package com.dy.dws.InteractionHosReviewWin;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.lang.Tuple;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dy.dws.function.MyAsyncDataStream;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsInteractionHosReviewBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.DateFormatUtil;
import com.dy.realtime.common.util.DorisMapFunction;
import com.dy.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.time.Duration;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.dws.InteractionHosReviewWin
 * @className: DwsInteractionHosReviewWin
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/16 19:43
 * @version: 1.0
 * 互动域医院粒度用户评价各窗口汇总表
 */

public class DwsInteractionHosReviewWin extends BaseApp {
    public static void main(String[] args) {
        new DwsInteractionHosReviewWin().start(Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW, Constant.TOPIC_DWD_USER_INTERACTION_HOS_REVIEW_WIN, 4, 10028);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<DwsInteractionHosReviewBean> etlStream = getEtlStream(dataStreamSource);
//        etlStream.print();
        SingleOutputStreamOperator<DwsInteractionHosReviewBean> hospitalId = getHospitalId(etlStream);
//        hospitalId.print();
        SingleOutputStreamOperator<DwsInteractionHosReviewBean> getHospitalName = getName(hospitalId);
//        getHospitalName.print();
        SingleOutputStreamOperator<DwsInteractionHosReviewBean> proccesStream = getProcces(getHospitalName);
//        proccesStream.print();
        SingleOutputStreamOperator<DwsInteractionHosReviewBean> result = getResult(proccesStream);
        result.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_hos_review_win"));
    }

    private static  SingleOutputStreamOperator<DwsInteractionHosReviewBean> getResult(SingleOutputStreamOperator<DwsInteractionHosReviewBean> proccesStream) {
        return proccesStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsInteractionHosReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<DwsInteractionHosReviewBean>() {
                    @Override
                    public long extractTimestamp(DwsInteractionHosReviewBean dwsInteractionHosReviewBean, long l) {
                        return dwsInteractionHosReviewBean.getTs();
                    }
                })).keyBy(new KeySelector<DwsInteractionHosReviewBean, String>() {
                    @Override
                    public String getKey(DwsInteractionHosReviewBean dwsInteractionHosReviewBean) throws Exception {
                        return dwsInteractionHosReviewBean.getHospitalName();
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<DwsInteractionHosReviewBean, DwsInteractionHosReviewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<DwsInteractionHosReviewBean, DwsInteractionHosReviewBean, String, TimeWindow>.Context context, Iterable<DwsInteractionHosReviewBean> iterable, Collector<DwsInteractionHosReviewBean> collector) throws Exception {
                        ArrayList<DwsInteractionHosReviewBean> dwsInteractionHosReviewBeans = ListUtil.toList(iterable);
                        long reviewNewUserCount = dwsInteractionHosReviewBeans.stream().mapToLong(e -> e.getReviewNewUserCount()).sum();
                        long reviewCount = dwsInteractionHosReviewBeans.stream().mapToLong(e -> e.getReviewCount()).sum();
                        long goodReviewCount = dwsInteractionHosReviewBeans.stream().mapToLong(e -> e.getGoodReviewCount()).sum();

                        String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());

                        DwsInteractionHosReviewBean next = dwsInteractionHosReviewBeans.iterator().next();
                        collector.collect(DwsInteractionHosReviewBean.builder()
                                .id(next.getId())
                                .createTime(next.getCreateTime())
                                .rating(next.getRating())
                                .doctorId(next.getDoctorId())
                                .hospitalId(next.getHospitalId())
                                .hospitalName(next.getHospitalName())
                                .patientId(next.getPatientId())
                                .userId(next.getUserId())
                                .stt(start)
                                .edt(end)
                                .curDate(curDate)
                                .goodReviewCount(goodReviewCount)
                                .reviewCount(reviewCount)
                                .reviewNewUserCount(reviewNewUserCount)
                                .build());
                    }
                });
    }

    private static SingleOutputStreamOperator<DwsInteractionHosReviewBean> getProcces(SingleOutputStreamOperator<DwsInteractionHosReviewBean> getHospitalName) {
        return getHospitalName.keyBy(new KeySelector<DwsInteractionHosReviewBean, Tuple2<String, String>>() {
            @Override
            public Tuple2 getKey(DwsInteractionHosReviewBean dwsInteractionHosReviewBean) throws Exception {
                return Tuple2.of(dwsInteractionHosReviewBean.getHospitalId(), dwsInteractionHosReviewBean.getUserId());
            }
        }).process(new KeyedProcessFunction<Tuple2<String, String>, DwsInteractionHosReviewBean, DwsInteractionHosReviewBean>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("isNew", String.class);
                valueState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(DwsInteractionHosReviewBean dwsInteractionHosReviewBean, KeyedProcessFunction<Tuple2<String, String>, DwsInteractionHosReviewBean, DwsInteractionHosReviewBean>.Context context, Collector<DwsInteractionHosReviewBean> collector) throws Exception {
                Long ts = dwsInteractionHosReviewBean.getTs();
                String s = DateFormatUtil.tsToDateTime(ts);
                String value = valueState.value();
                if (value == null) {
                    dwsInteractionHosReviewBean.setReviewNewUserCount(1L);
                    valueState.update(s);
                } else {
                    dwsInteractionHosReviewBean.setReviewNewUserCount(0L);
                }
                collector.collect(dwsInteractionHosReviewBean);
            }
        });
    }

    private static SingleOutputStreamOperator<DwsInteractionHosReviewBean> getName(SingleOutputStreamOperator<DwsInteractionHosReviewBean> hospitalId) {
        return AsyncDataStream.unorderedWait(hospitalId, new MyAsyncDataStream<DwsInteractionHosReviewBean>("dim_hospital") {
            @Override
            public void getResult(DwsInteractionHosReviewBean dwsInteractionHosReviewBean, JSONObject jsonObject) {
                dwsInteractionHosReviewBean.setHospitalName(jsonObject.getString("name"));
            }

            @Override
            public String getDimId(DwsInteractionHosReviewBean dwsInteractionHosReviewBean) {
                return dwsInteractionHosReviewBean.getHospitalId();
            }
        }, 100, TimeUnit.SECONDS, 20);
    }

    private static SingleOutputStreamOperator<DwsInteractionHosReviewBean> getHospitalId(SingleOutputStreamOperator<DwsInteractionHosReviewBean> etlStream) {
        return AsyncDataStream.unorderedWait(etlStream, new MyAsyncDataStream<DwsInteractionHosReviewBean>("dim_doctor") {
            @Override
            public void getResult(DwsInteractionHosReviewBean dwsInteractionHosReviewBean, JSONObject jsonObject) {
                dwsInteractionHosReviewBean.setHospitalId(jsonObject.getString("hospital_id"));
            }

            @Override
            public String getDimId(DwsInteractionHosReviewBean dwsInteractionHosReviewBean) {
                return dwsInteractionHosReviewBean.getDoctorId();
            }
        }, 100, TimeUnit.SECONDS, 20);
    }

    private static SingleOutputStreamOperator<DwsInteractionHosReviewBean> getEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, DwsInteractionHosReviewBean>() {
            @Override
            public void flatMap(String s, Collector<DwsInteractionHosReviewBean> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String create_time = jsonObject.getString("create_time");
                String rating = jsonObject.getString("rating");
                String review = jsonObject.getString("review");
                if (create_time != null && review != null && rating != null) {
                    DwsInteractionHosReviewBean dwsInteractionHosReviewBean = JSONObject.parseObject(jsonObject.toString(), DwsInteractionHosReviewBean.class);
                    dwsInteractionHosReviewBean.setReviewCount(1L);
                    Integer rating1 = dwsInteractionHosReviewBean.getRating();
                    if (rating1 >= 4) {
                        dwsInteractionHosReviewBean.setGoodReviewCount(1L);
                    } else {
                        dwsInteractionHosReviewBean.setGoodReviewCount(0L);
                    }
                    collector.collect(dwsInteractionHosReviewBean);
                }

            }
        });
    }
}
